package com.li.kafka.wechat.service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.li.kafka.wechat.conf.WechatTemplateProperties;
import com.li.kafka.wechat.utils.FileUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Optional;

@Slf4j
@Service
public class WechatTemplateServiceImpl implements WechatTemplateService {

  @Autowired private WechatTemplateProperties properties;

  @Autowired private Producer producer;

  @Override
  public WechatTemplateProperties.WechatTemplate getWechatTemplate() {
    List<WechatTemplateProperties.WechatTemplate> templates = properties.getTemplates();
    Optional<WechatTemplateProperties.WechatTemplate> first =
        templates.stream().filter((template) -> template.isActive()).findFirst();
    return first.isPresent() ? first.get() : null;
  }

  @Override
  public void templateReported(JSONObject reportInfo) {
    log.info("templateReported:[{}]", reportInfo);
    String topicName = "li-topic";
    String templateId = reportInfo.getString("templateId");
    JSONArray reportData = reportInfo.getJSONArray("result");
    ProducerRecord<String, Object> record = new ProducerRecord<>(topicName, templateId, reportData);
    producer.send(record);
  }

  @Override
  public JSONObject templateStatistics(String templateId) {
    if (properties.getTemplateResultType() == 0) {
      return FileUtils.readFile2JsonObject(properties.getTemplateResultFilePath()).get();
    }
    return null;
  }
}
